package com.atguigu.chapter11;

import com.atguigu.chapter5.source.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.junit.Test;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @ClassName: Flink09_Process_Timer
 * @Description:
 * @Author: kele
 * @Date: 2021/4/13 9:58
 *
 * https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/streaming/time_attributes.html#defining-in-ddl

 *  *
 * 获取事件时间
 * 注意：不管哪种方式添加事件时间，必须添加 水印
 *     1、通过流转换为表的时候添加事件时间
 *          方式一：添加新的字段作为事件时间
 *          方式二：将时间戳的字段作为事件时间
 *     2、在创建表的时候添加事件时间
 *          -- 创建表时添加的事件时间，是timestamp类型
 *          -- 将事件时间添加水印
 *
 **/
public class Flink10_Event_Timer_Table_SQL {

    /**
     *    在将流转化为表的时候设置事件时间
     *          添加新的字段方式
     */
    @Test
    public void Transfrom1(){

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        SingleOutputStreamOperator<WaterSensor> ds = env.fromElements(new WaterSensor("sensor_1", 1000L, 10),
                new WaterSensor("sensor_1", 2000L, 20),
                new WaterSensor("sensor_2", 3000L, 30),
                new WaterSensor("sensor_1", 4000L, 40),
                new WaterSensor("sensor_1", 5000L, 50),
                new WaterSensor("sensor_2", 6000L, 60))
//                添加水印
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                                .withTimestampAssigner((element, recordTimestamp) -> element.getTs())
                );

        Table table = tenv.fromDataStream(ds, $("id"), $("ts"), $("vc"), $("ets").rowtime());

        table.execute().print();
    }


    /**
     *    在将流转化为表的时候设置事件时间
     *         将已经存在的字段作为事件时间（stream中也必须添加水印）
     */
    @Test
    public void Transfrom2(){

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        SingleOutputStreamOperator<WaterSensor> ds = env.fromElements(new WaterSensor("sensor_1", 1000L, 10),
                new WaterSensor("sensor_1", 2000L, 20),
                new WaterSensor("sensor_2", 3000L, 30),
                new WaterSensor("sensor_1", 4000L, 40),
                new WaterSensor("sensor_1", 5000L, 50),
                new WaterSensor("sensor_2", 6000L, 60))
//                添加水印
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                                .withTimestampAssigner((element, recordTimestamp) -> element.getTs())
                );

        Table table = tenv.fromDataStream(ds, $("id"), $("ts").rowtime(), $("vc"));

        table.execute().print();


    }


    /**
     * 创建表的时候添加事件时间
     *
     */
    @Test
    public void CtreateTbale(){

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        tenv.executeSql("create table sen(\n" +
                " id string,\n" +
                " ts bigint,\n" +
                " vc int,\n" +
//                设置事件时间
                " ets as to_timestamp( from_unixtime( ts,'yyyy-MM-dd HH:mm:ss') ),\n" +
//                设置水印
                " watermark for ets as ets - interval '5' second\n" +
                " )with(\n" +
                " 'connector' = 'filesystem',\n" +
                " 'path' = 'in/sensor.txt',\n" +
                " 'format' = 'csv'\n" +
                ")");

        tenv.sqlQuery("select * from sen").execute().print();

      //  Table table = tenv.from("sen");

     //   table.execute().print();
    }

}
